package com.smsc.headend.connector.aop;

import com.smsc.headend.connector.annotations.DeviceSignInLimiter;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import io.micrometer.core.instrument.Counter;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.stereotype.Component;

import java.lang.reflect.Method;


@Slf4j
@Component
@Aspect
public class KafkaSendAspect {


    @Autowired
    @Qualifier("taskOutboundCounter")
    Counter taskOutboundCounter;

    @Pointcut("execution (* org.springframework.kafka.core.KafkaTemplate.send(String,..))")
    public void kafkaSend() {
    }


    @Around("kafkaSend()")
    public Object kafkaSendAround(ProceedingJoinPoint joinPoint) {
        try {
            MethodSignature signature = (MethodSignature) joinPoint.getSignature();
            Method method = signature.getMethod();
            String topic = (String) joinPoint.getArgs()[0];
            if (topic.equals(TaskKafkaTopic.TASK_SEND_TO_UTE)) {
                log.debug("count for task");
                taskOutboundCounter.increment();
            }
            Object proceed = joinPoint.proceed();
            return proceed;
        } catch (Throwable e) {
            log.error("errorIn:{}", joinPoint.getSignature(), e);
        }
        return null;
    }

}
